package com.hu.api

import com.hu.entity.SensorReading
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

/**
 * @Author: hujianjun
 * @Create Date: 2020/11/25 11:25
 * @Describe: 测试Table API
 */
object TableAPIDemo {
  def main(args: Array[String]): Unit = {
    val parameterTool = ParameterTool.fromArgs(args)
    val inputPath = parameterTool.get("input")

    // 1.获取使用的Table计划期，默认使用blink的计划器
    val fsSettings = EnvironmentSettings.newInstance().inStreamingMode().useOldPlanner().build()

    // 2.获取流执行环境
    val envStream = StreamExecutionEnvironment.getExecutionEnvironment
    envStream.setParallelism(1)

    // 3.获取表的流执行环境
    val tableStreamEnv = StreamTableEnvironment.create(envStream, fsSettings)

    // 4.数据操作
    val dataStream = envStream.readTextFile(inputPath)

    val sensorStream = dataStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //4.1 table操作
    //用dataStream 注册一个视图
    tableStreamEnv.createTemporaryView("t_sensor", sensorStream)
    val selectResult = tableStreamEnv.from("t_sensor")
      .filter('id === "sensor_1")
      .select('id, 'address as "city", 'ts, 'temperature)
    selectResult.printSchema()
    selectResult.toRetractStream[Row].print("使用table api查询表数据")

    //4.2 使用sql查询
    val sqlResult = tableStreamEnv.sqlQuery(
      """
        |select id,sum(temperature) temp_total
        |from t_sensor where id='sensor_1'
        |group by id
        |""".stripMargin)

    sqlResult.toRetractStream[Row].print("使用sql查询表数据")

    // 4.3 使用connect构建表
    //    tableStreamEnv.connect(new FileSystem().path("D:\\hujianjun\\I\\iyb-code\\hu-learn-code\\flink-scala-learn\\src\\main\\resources\\data\\sensor.csv"))
    //      .withFormat(new Csv().fieldDelimiter(',').deriveSchema())
    //      .withSchema(new Schema().field("id", DataTypes.STRING())
    //        .field("address", DataTypes.STRING())
    //        .field("ts", DataTypes.BIGINT())
    //        .field("temperature", DataTypes.DOUBLE())
    //      )
    //      .createTemporaryTable("t_sensor_result")

    //将查询结果插入到使用conn构建的结果表中
    //    selectResult.executeInsert("t_sensor_result")


    // 5.数据sink
    //    sensorStream.print()

    // 6.执行
    envStream.execute("Table API test")
  }
}
